package com.nokia.pc3.core.format.input;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;

import com.nokia.pc3.core.xform.Transform;
import com.nokia.pc3.core.xform.TransformFactory;

/**
 * @author jcroke
 * 
 */
public class NokiaUnstructuredRecordInputFormat extends
		FileInputFormat<LongWritable, NokiaXFormRecord> {

	@Override
	public RecordReader<LongWritable, NokiaXFormRecord> createRecordReader(
			InputSplit arg0, TaskAttemptContext arg1) throws IOException,
			InterruptedException {
		com.nokia.pc3.core.dac.DAC dac = null;
		Configuration conf = arg1.getConfiguration();
		Class<?> DACInit = conf.getClass("pc3.core.DACInit",
				com.nokia.pc3.core.dac.DACInit.class);
		try {
			dac = ((com.nokia.pc3.core.dac.DACInit) DACInit
					.getDeclaredConstructor().newInstance()).getDAC();
		} catch (Exception e) {
			e.printStackTrace();
		}
		Transform transform = null;
		TransformFactory factory = new TransformFactory(dac);
		try {
			transform = factory
					.getTransform("UnstructuredNokiaRecordInputFileFormat");
		} catch (ClassNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		// TextInputFormat also just makes a LineRecordReader, so we do so here
		// and give it to the NokiaRecordReader
		RecordReader<LongWritable, Text> textreader = new LineRecordReader();
		RecordReader<LongWritable, NokiaXFormRecord> result = new NokiaUnstructuredRecordReader(
				textreader, transform);
		return result;
	}

	@Override
	protected boolean isSplitable(JobContext context, Path file) {
		CompressionCodec codec = new CompressionCodecFactory(
				context.getConfiguration()).getCodec(file);
		return codec == null;
	}
}
